package com.idanchuang.count.core.config;

import com.alibaba.fastjson.JSON;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.idanchuang.count.core.deserializer.SimpleCountIncrEventDeserializer;
import com.idanchuang.count.core.entity.SimpleCountIncrEvent;
import com.idanchuang.count.support.KafkaAsyncSender;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author Richard_yyf
 * @version 1.0 2021/5/27
 */
@Configuration
@EnableKafka
@EnableConfigurationProperties(KafkaProperties.class)
@Slf4j
public class KafkaConfiguration {

    // ====================  消费者配置  ====================

    private final KafkaProperties properties;

    /**
     * 消费并发 默认1
     */
    @Value("${count.async.concurrency:1}")
    private Integer concurrency;

    @Value("${count.simple.concurrency:3}")
    private Integer simpleCountConcurrency;

    @Value("${count.simple.enable:true}")
    private boolean simpleCountEnable;

    public KafkaConfiguration(KafkaProperties properties) {
        this.properties = properties;
    }

    public ConsumerFactory<Object, Object> userCountEventConsumerFactory() {
        Map<String, Object> config = properties.buildConsumerProperties();
        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
    customContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(userCountEventConsumerFactory());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.getContainerProperties().setPollTimeout(properties.getListener().getPollTimeout().toMillis());
        factory.setBatchListener(Boolean.TRUE);
        // 本地测试阶段先不开启消费
        factory.setAutoStartup(true);

        return factory;
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
    customSingleContainerFactory() {
        // 单笔
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(userCountEventConsumerFactory());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.getContainerProperties().setPollTimeout(properties.getListener().getPollTimeout().toMillis());
        factory.setBatchListener(Boolean.FALSE);
        factory.setAutoStartup(true);

        return factory;
    }


    // =========================   简单计数场景   ====================================

    public ConsumerFactory<Object, Object> simpleCountEventConsumerFactory() {
        Map<String, Object> config = properties.buildConsumerProperties();
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SimpleCountIncrEventDeserializer.class.getName());
        return new DefaultKafkaConsumerFactory<>(config);
    }


    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, SimpleCountIncrEvent>>
    simpleCountBatchConsumeContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, SimpleCountIncrEvent> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(simpleCountEventConsumerFactory());
        factory.setConcurrency(simpleCountConcurrency);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.getContainerProperties().setPollTimeout(properties.getListener().getPollTimeout().toMillis());
        factory.setBatchListener(Boolean.TRUE);
        factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
        // 部分环境不开启
        factory.setAutoStartup(simpleCountEnable);

        return factory;
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, SimpleCountIncrEvent>>
    simpleCountSingleConsumeContainerFactory() {
        // 单笔
        ConcurrentKafkaListenerContainerFactory<String, SimpleCountIncrEvent> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(simpleCountEventConsumerFactory());
        factory.setConcurrency(simpleCountConcurrency);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getContainerProperties().setPollTimeout(properties.getListener().getPollTimeout().toMillis());
        factory.setBatchListener(Boolean.FALSE);
        factory.setErrorHandler(new SeekToCurrentErrorHandler(6));
        // 部分环境不开启
        factory.setAutoStartup(simpleCountEnable);

        return factory;
    }


    // ====================  生产者配置  ====================

    @Bean
    public ProducerFactory<Object, Object> producerFactory() {
        Map<String, Object> config = properties.buildProducerProperties();
        // ...
        log.info("kafka producer configs: {}", JSON.toJSONString(config, true));
        return new DefaultKafkaProducerFactory<>(config);
    }


    @Bean
    public KafkaTemplate<Object, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    @SuppressWarnings("UnstableApiUsage")
    public EventBus kafkaSendEventBus() {
        final ExecutorService eventExecutor = new ThreadPoolExecutor(1, 1, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100000),
                new ThreadFactoryBuilder().setNameFormat("kafka-send-worker-%d").build());

        return new AsyncEventBus("kafkaSendEventBus", eventExecutor);
    }


    @Bean
    @SuppressWarnings("UnstableApiUsage")
    public KafkaAsyncSender kafkaAsyncSender(EventBus kafkaSendEventBus, KafkaTemplate<Object, Object> kafkaTemplate) {
        KafkaAsyncSender kafkaAsyncSender = new KafkaAsyncSender(kafkaTemplate);
        kafkaSendEventBus.register(kafkaAsyncSender);
        return kafkaAsyncSender;
    }


}
